using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using CassandraHost = Me.Prettyprint.Cassandra.Service.CassandraHost;
using CassandraHostConfigurator = Me.Prettyprint.Cassandra.Service.CassandraHostConfigurator;
using Me.Prettyprint.Cassandra.Utils;

namespace Me.Prettyprint.Cassandra.Connection
{
    /// <summary>
    /// HostTimeoutTracker is what it sounds like. It will retry a host that is unresponsive.
    /// If it remains unresponsive, the host will eventually be marked as down. Once a host
    /// is marked down, it is passed to the host retry service to detect when it returns to
    /// normal operation.
    /// </summary>
	public class HostTimeoutTracker : BackgroundCassandraHostService
	{
		private static readonly Logger logger = new Logger(typeof(HostTimeoutTracker));
		private ConcurrentDictionary<CassandraHost, ConcurrentQueue<long>> timeouts;
		private ConcurrentDictionary<CassandraHost, long> suspended;
		private int timeoutCounter;
		private int timeoutWindow;
		private int nodeSuspensionDurationInSeconds;
        private CancellationTokenSource tokenSource = new CancellationTokenSource();
		
		public const int DEF_TIMEOUT_COUNTER = 10;
		public const int DEF_TIMEOUT_WINDOW = 500;
		public const int DEF_NODE_SUSPENSION_DURATION_IN_SECONDS = 10;
		public const int DEF_NODE_UNSUSPEND_CHECK_DELAY_IN_SECONDS = 10;
		
		public HostTimeoutTracker(HConnectionManager connectionManager, CassandraHostConfigurator cassandraHostConfigurator):base(connectionManager, cassandraHostConfigurator)
		{
			retryDelayInSeconds = cassandraHostConfigurator.HostTimeoutUnsuspendCheckDelay;
			timeouts = new ConcurrentDictionary<CassandraHost, ConcurrentQueue<long>>();
			suspended = new ConcurrentDictionary<CassandraHost, long>();
			timeoutCounter = cassandraHostConfigurator.HostTimeoutCounter;
			timeoutWindow = cassandraHostConfigurator.HostTimeoutWindow;
			nodeSuspensionDurationInSeconds = cassandraHostConfigurator.HostTimeoutSuspensionDurationInSeconds;
            var token = tokenSource.Token;
            backgroundTask = Task.Factory.StartNew(() => StartBackgroundProcess(token));
		}
        public void StartBackgroundProcess(CancellationToken token)
        {
            if (token.IsCancellationRequested)
            {
                token.ThrowIfCancellationRequested();
            }
            while (!token.IsCancellationRequested)
            {
                Thread.Sleep(retryDelayInSeconds * 1000);
                Unsuspend();
            }
            if (token.IsCancellationRequested)
            {
                token.ThrowIfCancellationRequested();
            }
        }
		public bool CheckTimeout(CassandraHost cassandraHost)
		{
            if (!timeouts.ContainsKey(cassandraHost))
            {
                timeouts.AddOrUpdate(cassandraHost, new ConcurrentQueue<long>(),null);
            }
			long currentTimeMillis = (System.DateTime.Now.Ticks - 621355968000000000) / 10000;
			timeouts[cassandraHost].Enqueue(currentTimeMillis);
			bool timeout = false;
			// if there are 3 timeouts within 500ms, return false
			if (timeouts[cassandraHost].Count > timeoutCounter)
			{
                System.Int64 last = 0;
                timeouts[cassandraHost].TryDequeue(out last);
				if (last < (currentTimeMillis - timeoutWindow))
				{
					timeout = true;
					connectionManager.SuspendCassandraHost(cassandraHost);
                    if (!suspended.ContainsKey(cassandraHost))
                    {
                        suspended.AddOrUpdate(cassandraHost, currentTimeMillis, null);
                    }
				}
			}
			return timeout;
		}
        public void Unsuspend()
        {
            IEnumerator<CassandraHost> keysEnumerator = suspended.Keys.GetEnumerator();
            while (keysEnumerator.MoveNext())
            {
                long time = 0;
                CassandraHost host = keysEnumerator.Current;
                suspended.TryGetValue(host, out time);
                if (time < (DateTime.Now.Ticks - (nodeSuspensionDurationInSeconds * 1000000000)))
                {
                    connectionManager.UnsuspendCassandraHost(host);
                    long val = 0;
                    suspended.TryRemove(host, out val);
                }
            }
        }
        /// <summary>
        /// Not implemented in Hector, so nectar also leaves it blank
        /// </summary>
		public override void ApplyRetryDelay()
		{
		}

        internal override void Shutdown()
		{
			logger.Info("Shutting down HostTimeoutTracker");
			if (backgroundTask != null)
            {
                tokenSource.Cancel();
            }
			logger.Info("HostTimeTracker shutdown complete.");
		}
	}
}